package com.atguigu.sparkstreaming.commitoffsets

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/8/22
 *
 *
 *    演示偏移量提交到kakfa代码的编写位置
 *
 *
 *
 *
 * Caused by: java.io.NotSerializableException:
 *        Object of org.apache.spark.streaming.kafka010.DirectKafkaInputDStream is being serialized  possibly
 *        as a part of closure of an RDD operation.
 *
 *          DirectKafkaInputDStream 作为了RDD算子操作闭包的一部分，需要序列化。
 *
 *        This is because  the DStream object is being referred to from within the closure.
 *        Please rewrite the RDD operation inside this DStream to avoid this.
 *        This has been enforced to avoid bloating of Spark tasks  with unnecessary objects.
 *
 *            让你去重写RDD的算子，避免出现闭包！
 *
 */
object CommitOffsetsToKafkaDemo3 {

  def main(args: Array[String]): Unit = {


    val streamingContext = new StreamingContext("local[*]", "wordcount", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "sz220409test",
      "auto.offset.reset" -> "latest",
      //①取消自动提交
      "enable.auto.commit" -> "false"
    )


    val topics = Array("topicD")


    // 初始DS，才是DirectKafkaInputDStream
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    ds.foreachRDD(rdd => {

      //②获取到当前批次的偏移量
      val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      rdd.foreach(record => {

        //打印输出
        println(Thread.currentThread().getName + record.value())

        //③调用API，把每个分区消费到的utilOffset提交到kafka集群的_comsumer_offsets中
        //错误的示范，产生闭包
        //ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)
      })

      //③调用API，把每个分区消费到的utilOffset提交到kafka集群的_comsumer_offsets中
      //从原理上来说，偏移量在Driver端获取，一定在Driver端进行提交，提交的位置只能是RDD.算子的外面！
      ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)

    })


    streamingContext.start()

    streamingContext.awaitTermination()

  }

}
